package zjs.dc.controller;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

/**
 * @Class:
 * @Author: xyl
 * @Description: 针对企查查 t_tax_illegal_detail 表和中数 t_tax_punish 表进行数据映射开发
 * @Date: 2021/12/30 16:51
 */
public class TTaxIllegalDetail {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder().appName("t_tax_illegal_detail").enableHiveSupport().getOrCreate();
        spark.sparkContext().setLogLevel("ERROR");
        Dataset<Row> zs_data = spark.sql("select * from a_dc_ep_ods.zs_t_tax_punish");
        long zs_count = zs_data.count();
        if(zs_count>0){
            /**
             * t_tax_illegal_detail 税收违法详情
             */
            //获取中数删除数据表中的数据 --- 中数删除逻辑：公司的任何一条数据发生变化，先删除这个公司相关的该表的所有数据
            Dataset<Row> delete_data = spark.sql("select \n" +
                    "a.id,a.key_no,a.company_id,a.company_name,a.tax_illegal_id,a.taxpayer_id,a.taxpayer_name,a.taxpayer_number,a.org_code,a.address, \n" +
                    "a.illegal_time,a.publish_time,a.case_nature,a.tax_gov,a.illegal_content, \n" +
                    "a.oper_id,a.oper_name,a.oper_cer_no,a.oper_gender,a.oper_cer_type, \n" +
                    "a.finance_chief_id,a.finance_chief_cer_no,a.finance_chief_gender,a.finance_chief_name,a.finance_chief_cer_type, \n" +
                    "a.agency_person_id,a.agency_company_name,a.agency_key_no,a.agency_person_name,a.agency_person_cer_type,a.agency_person_cer_no,a.agency_person_gender, \n" +
                    "c.jobid,'-1' as isadd  \n" +
                    "from (select * from a_dc_ep_dwi.t_tax_illegal_detail where company_id is not null and company_id !='')a  \n" +
                    "inner join (select * from a_dc_ep_ods.t_eci_company where credit_code is not null and credit_code !='') b on a.company_id=b.company_id \n" +
                    "inner join (select * from a_dc_ep_ods.zs_t_tax_punish_del where creditcode is not null and creditcode !='')c on upper(trim(b.credit_code)) = upper(trim(c.creditcode))\n");
            delete_data.createOrReplaceTempView("tmp_t_tax_illegal_detail");
            spark.sql("insert overwrite table a_dc_ep_incr.t_tax_illegal_detail select * from tmp_t_tax_illegal_detail");

            //获取中数的 更新+ 新数据
            Dataset<Row> update_data = spark.sql("SELECT " +
                    "       md5(concat(c.company_id,a.RECORD_ID)),\n" +
                    "       c.key_no,\n" +
                    "       c.company_id,\n" +
                    "       c.company_name,\n" +
                    "       \"\",\n" +
                    "       \"\",\n" +
                    "       a.TAX_NAME,\n" +
                    "       a.TAX_REGNO,\n" +
                    "       d.org_no,\n" +
                    "       c.address,\n" +
                    "       \"\",\n" +
                    "       \"\",\n" +
                    "       a.CASE_TYPE,\n" +
                    "       if(instr(split(a.PUNISH_DETAIL,\"检查\")[0],\"税务局\")>0 and length(split(a.PUNISH_DETAIL,\"检查\")[0])<255,split(a.PUNISH_DETAIL,\"检查\")[0],\"\"),\n" +
                    "       a.PUNISH_DETAIL,\n" +
                    "       \"\",\n" +
                    "       trim(case when a.LEG_DETAIL != \"\" then split(a.LEG_DETAIL,\" \")[0] else \"\" end),\n" +
                    "       trim(case when a.LEG_DETAIL != \"\" then split(a.LEG_DETAIL,\" \")[3] else \"\" end),\n" +
                    "       trim(case when a.LEG_DETAIL != \"\" then split(a.LEG_DETAIL,\" \")[1] else \"\" end),\n" +
                    "       trim(case when a.LEG_DETAIL != \"\" then split(a.LEG_DETAIL,\" \")[2] else \"\" end),\n" +
                    "       \"\",\n" +
                    "       trim(case when (a.FINA_DETAIL != \"\" AND a.FINA_DETAIL != \"无\") then split(a.FINA_DETAIL,\" \")[3] else \"\" end),\n" +
                    "       trim(case when (a.FINA_DETAIL != \"\" AND a.FINA_DETAIL != \"无\") then split(a.FINA_DETAIL,\" \")[1] else \"\" end),\n" +
                    "       trim(case when (a.FINA_DETAIL != \"\" AND a.FINA_DETAIL != \"无\") then split(a.FINA_DETAIL,\" \")[0] else \"\" end),\n" +
                    "       trim(case when (a.FINA_DETAIL != \"\" AND a.FINA_DETAIL != \"无\") then split(a.FINA_DETAIL,\" \")[2] else \"\" end),\n" +
                    "       \"\",\n" +
                    "       trim(case when (size(split(a.AGE_DETAIL,\"，\")) > 2) then split(a.AGE_DETAIL,\"，\")[0] else \"\" end),\n" +
                    "       \"\",\n" +
                    "       trim(case when (size(split(a.AGE_DETAIL,\"，\")) > 2) then split(a.AGE_DETAIL,\"，\")[1] else \"\" end),\n" +
                    "       case when (size(split(a.AGE_DETAIL,\"，\")) > 2) then \"身份证\" else \"\" end,\n" +
                    "       trim(case when (size(split(a.AGE_DETAIL,\"，\")) > 2) then split(a.AGE_DETAIL,\"，\")[2] else \"\" end),\n" +
                    "       trim(case when (size(split(a.AGE_DETAIL,\"，\")) > 2) then if((cast(substring(split(a.AGE_DETAIL,\"，\")[2],-2,1) as INT) % 2) == 0, \"女\",\"男\")  else \"\" end),\n" +
                    "       a.JOBID,\n" +
                    "       \"0\" as isadd  " +
                    "FROM \n" +
                    "  (SELECT * FROM a_dc_ep_ods.zs_t_tax_punish WHERE ENTID IS NOT NULL AND ENTID !='') a \n" +
                    "INNER JOIN \n" +
                    "  (SELECT * FROM a_dc_ep_ods.zs_t_tax_punish_del WHERE creditcode IS NOT NULL AND creditcode!='') b \n" +
                    "ON a.entid = b.entid \n" +
                    "INNER JOIN \n" +
                    "  (SELECT * FROM a_dc_ep_ods.t_eci_company WHERE credit_code IS NOT NULL AND credit_code!='') c \n" +
                    "ON upper(trim(b.creditcode)) = upper(trim(c.credit_code)) \n" +
                    "left JOIN \n" +
                    "  (SELECT * FROM a_dc_ep_dwi.t_eci_companyattr WHERE company_id IS NOT NULL AND company_id!='') d \n" +
                    "ON c.company_id = d.company_id \n" +
                    "where b.creditcode is not null and b.creditcode !='' and c.credit_code is not null and c.credit_code!='' ");

            update_data.createOrReplaceTempView("tmp_t_tax_illegal_detail1");
            spark.sql("insert into table a_dc_ep_incr.t_tax_illegal_detail select * from tmp_t_tax_illegal_detail1");
        }else {
            spark.sql("TRUNCATE TABLE a_dc_ep_incr.t_tax_illegal_detail");
            System.out.println("中数本期数据涉及该表的数据为空。。。。。。");
        }


        spark.stop();
    }

}